package org.yonggan.shop.sql.report

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.expressions.{Window, WindowSpec}
import org.yonggan.shop.constant.ConfigurationManager
import org.yonggan.shop.utils.SparkUtils

/**
  * 各个分类（手机、书）商品页面被浏览次数最多的TON2
  */
object RptGroupedTopNDSL {

  def main(args: Array[String]): Unit = {

    val spark = SparkUtils.getSparkSession("商品页面被浏览次数最多的TON2")

    spark.sparkContext.setLogLevel("WARN")

    // 文件输出
    val jsonDF: DataFrame = spark.read.json(ConfigurationManager.TASK_INJSON)
    import org.apache.spark.sql.functions._

    val bsDF: DataFrame = jsonDF.filter(
      """
        | event_type = 1
        | and cid is not null
        | and pid is not null
        | and page = "/pages/content/detail/detail"
      """.stripMargin) //
      .groupBy("cid","pid").agg(count("*") as "ctn")
      .select("cid","pid", "ctn")

    // 分组topN
    val spec: WindowSpec = Window.partitionBy("cid").orderBy(desc("ctn"))
    val top2 = bsDF.withColumn("rn", row_number.over(spec))
      .where("rn <= 2").drop("rn")

//    +---+--------------------+---+
//    |cid|                 pid|ctn|
//    +---+--------------------+---+
//    |  1|5ace32e357f58a368...| 56|
//    |  1|5ace2f8157f58a368...| 55|
//    |  4|5ace2f3c57f58a368...|190|
//    |  4|5ad16c9a57f58a330...| 59|
//    +---+--------------------+---+

    top2.show()

  }

}
